package boot

import (
	"context"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/tx7do/kratos-transport/broker"
	"github.com/tx7do/kratos-transport/broker/kafka"
	"github.com/tx7do/kratos-transport/broker/nsq"
	"kratos_cron_task/internal/conf"
)

type Broker struct {
	conf   *conf.Bootstrap
	logger *log.Helper
	//Kafka  broker.Broker
	//Nsq broker.Broker
}

func NewBroker(conf *conf.Bootstrap, logger log.Logger) *Broker {
	return &Broker{
		conf:   conf,
		logger: log.NewHelper(logger),
		//Kafka:  NewBootKafka(conf, logger).Run(),
		//Nsq: NewBootNsq(conf, logger).Run(),
	}
}
func NewBootNsq(conf *conf.Bootstrap, logger log.Logger) *Nsq {
	return &Nsq{
		conf,
		log.NewHelper(logger),
	}
}

func NewBootKafka(conf *conf.Bootstrap, logger log.Logger) *Kafka {
	return &Kafka{
		conf,
		log.NewHelper(logger),
	}
}

type Kafka struct {
	conf   *conf.Bootstrap
	logger *log.Helper
}

func (b *Kafka) Run() broker.Broker {
	broker := kafka.NewBroker(
		broker.WithAddress(b.conf.MessageQueue.Kafka.Broker.Addr...),
	)
	return broker
}

type Nsq struct {
	conf   *conf.Bootstrap
	logger *log.Helper
}

func (b *Nsq) Run() broker.Broker {
	ctx := context.Background()
	broker := nsq.NewBroker(
		broker.WithOptionContext(ctx),
		broker.WithAddress(b.conf.MessageQueue.Nsq.Broker.Addr...),
	)
	_ = broker.Init()
	if err := broker.Connect(); err != nil {
		b.logger.WithContext(context.Background()).Errorw("reason", "boot nsq connect error", "error", err)
	}
	return broker
}
